package com.etlmaster.executor.steps.sql;

import com.etlmaster.core.DataSourceHolder;
import com.etlmaster.executor.utils.DataCollectionUtils;
import com.etlmaster.jdbc.JdbcTemplate;

/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
public abstract class SqlProcessor {

	protected SqlStepMeta meta;

	public boolean process() {
		boolean flag = true;
		try {
			JdbcTemplate jdbcTemplate = new JdbcTemplate(meta.getDsName());

			String dsType = DataCollectionUtils.getDsType(meta.getDsName());

			if (meta.getSql() != null && meta.getSql().length() > 0) {
				if (meta.getIsOneSql().equalsIgnoreCase("false")) {
					String[] sqls = meta.getSql().trim().split("(?<!\\\\);");
					for (String sql : sqls) {
						if (sql.length() > 0) {
							meta.log("INFO", "Sql语句：{}", sql);
							try {
								if (dsType.equalsIgnoreCase("db2")) {
									sql = sql.replace("\\;", ";");
								}
								runSql(sql, jdbcTemplate);
							} catch (Exception e) {
								if (sql.toLowerCase().contains("drop")) {
									meta.log("ERROR",
											"drop sql {}失败,继续执行", sql);
									meta.log("ERROR", "{}",
											e.getMessage());
									continue;
								} else {
									throw e;
								}
							}
						}
					}
				} else {
					runSql(meta.getSql(), jdbcTemplate);

				}
			} else {
				flag = false;
				meta.log("WARN", "配置的SQL语句为空，无法执行");
			}
		} catch (Exception e) {
			meta.log("ERROR", e.getMessage());
			flag = false;
		}
		return flag;
	}

	protected void runSql(String sql, JdbcTemplate jdbcTemplate) {
		int row = jdbcTemplate.update(sql);
		meta.getStepState().getReturnValue().put("row_count", row);
		meta.log("INFO", "Sql执行成功，影响行数：{},数据源为:{},语句：{}", row + "",
				meta.getDsName(), sql);
	}

	public void setMeta(SqlStepMeta meta) {
		this.meta = meta;
	}

	public static SqlProcessor getSqlProcessor(SqlStepMeta meta, SqlStep step) {
		String dsType = DataSourceHolder.getInstance().getDataSourceType(
				meta.getDsName());
		SqlProcessor sqlProcessor = null;
		if (dsType.equalsIgnoreCase("hive")) {
			sqlProcessor = new HiveSqlProcessor();
		} else {
			sqlProcessor = new RdbSqlProcessor();
		}
		sqlProcessor.setMeta(meta);
		return sqlProcessor;
	}
}
